package cluster

import (
	"context"

	"gddgame.cc/galaxy/asteroid"
	"gddgame.cc/galaxy/utils/serialize"

	"fmt"
	"sync"

	"gddgame.cc/galaxy/utils/def"
)

type HashService interface {
	Service
}

// 通过token进行一致性hash分布式
// 自动生成服务
type HashCenter struct {
	container *container
	manages   sync.Map

	result sync.Pool
}

type hash struct {
	def.Logger
	*asteroid.Room
	HashService

	ctx      context.Context
	cancelFn func()

	center   *HashCenter
	token    string
	instance interface{}

	init bool
}

var (
	hashServices = make(map[string]HashService)

	//hashMessagePool = &sync.Pool{}
	//hashMutex       = sync.RWMutex{}
)

func getHashService(module string) HashService {
	//hashMutex.RLock()
	//defer hashMutex.RUnlock()
	if mm, ok := hashServices[module]; ok {
		return mm
	}
	// 需要预先register
	panic(fmt.Errorf("manager is miss:%s，need register first", module))
}

func registerHash(service HashService) {
	//hashMutex.Lock()
	//defer hashMutex.Unlock()
	hashServices[service.ID()] = service
}
func init() {
	//hashMessagePool.New = func() interface{} {
	//	return &hashMessage{}
	//}
}

func newHash(container *container) *HashCenter {
	mutex := sync.Mutex{}
	m := &HashCenter{
		container: container,
		manages:   sync.Map{},
	}
	container.global.On("hash/create", func(message asteroid.Message) {
		bag := serialize.GetBag()
		bag.SetList(message.List())
		module := bag.ReadString(0)
		token := bag.ReadString(1)
		bag.Release()
		identity := token
		manager := m.LoadManager(module)
		_, ok := manager.LoadService(identity)
		if !ok {
			mutex.Lock()
			_, ok := manager.LoadService(identity)
			if !ok {
				m.getService(module, token)
			}
			mutex.Unlock()
		}
		message.Response(true)
	})
	container.global.On("hash/exec", func(message asteroid.Message) {
		bag := serialize.GetBag()
		bag.SetList(message.List())
		module := bag.ReadString(0)
		token := bag.ReadString(1)
		event := bag.ReadInterface(2)
		params := bag.ReadInterface(3)
		bag.Release()
		var instance *hash
		identity := token
		manager := m.LoadManager(module)
		i, ok := manager.LoadService(identity)
		if !ok {
			mutex.Lock()
			i, ok := manager.LoadService(identity)
			if !ok {
				instance = m.getService(module, token)
			} else {
				instance = i.(*hash)
			}
			mutex.Unlock()
		} else {
			instance = i.(*hash)
		}
		instance.LocalEventAsync(event, params, message.GetFrom(), message.GetTarget())
		message.Response(true)
	})
	container.global.On("hash/delete", func(message asteroid.Message) {
		bag := serialize.GetBag()
		bag.SetList(message.List())
		module := bag.ReadString(0)
		token := bag.ReadString(1)
		bag.Release()
		identity := token
		manager := m.LoadManager(module)
		client, ok := manager.LoadClient(identity)
		if ok {
			client.(*Client).close()
			manager.DeleteClient(identity)
		}
		i, ok := manager.LoadService(identity)
		if ok {
			i.(*hash).close()
			manager.DeleteService(identity)
		}
		message.Response(true)
	})
	m.result.New = func() interface{} {
		return make(chan string)
	}
	//m.cluster.Run()
	return m
}

func (m *HashCenter) LoadManager(module string) *ServiceManager {
	manager, ok := m.manages.Load(module)
	if !ok {
		manager = NewManager(getHashService(module))
		m.manages.Store(module, manager)
	}
	return manager.(*ServiceManager)
}

func (m *HashCenter) getService(module string, token string) *hash {
	identity := token
	var tr *asteroid.Room
	var hashInstance *hash

	manager := m.LoadManager(module)
	service := manager.service.(HashService)

	tr, _ = m.container.region.OpenRoom(module + "/" + identity)

	ctx, cancelFn := context.WithCancel(m.container.engine.ctx)
	hashInstance = &hash{
		HashService: service,
		Room:        tr,
		center:      m,
		token:       token,
		init:        false,

		ctx:      ctx,
		cancelFn: cancelFn,
	}
	hashInstance.Logger = tr.LoggerWrap("Cluster", module)
	// 保证流程
	tr.On(JoinEvent, func(message asteroid.Message) {
		message.Response(true)
	})
	tr.On(LeaveEvent, func(message asteroid.Message) {
		message.Response(true)
	})
	service.Init(hashInstance)
	hashInstance.init = true
	manager.StoreService(identity, hashInstance)
	return hashInstance
}

func (m *HashCenter) OpenService(module string, token string) string {
	identity := module + "/" + token
	target := m.result.Get().(chan string)
	defer m.result.Put(target)
	node := m.container.region.Alloc(identity)
	m.container.global.Direct(node).List(module, token).CallbackTo(func(success bool, data interface{}) {
		if success {
			target <- node
		} else {
			target <- ""
		}
	}).Send("hash/create")
	return <-target
}

func (m *HashCenter) GetClient(module string, token string) *Client {
	identity := token
	var client *Client
	var node string
	var tr *asteroid.Room
	manager := m.LoadManager(module)
	i, ok := manager.LoadClient(identity)
	if ok {
		client = i.(*Client)
		goto Return
	}
	tr, _ = m.container.region.OpenRoom(module + "/" + identity)
	node = m.container.region.Alloc(identity)
	client = newClient(m.container, identity, node, m, manager.service, tr)
	manager.service.InitClient(client)
	manager.StoreClient(identity, client)
Return:
	return client
}

func (m *HashCenter) closeClient(client *Client, service Service) {
	service.CloseClient(client)
	manager := m.LoadManager(client.service.ID())
	manager.DeleteClient(client.identity)
}

func (i *hash) close() {
	i.CloseInstance(i)
	i.cancelFn()
}

func (i *hash) Close() error {
	i.center.container.global.All().List(i.HashService.ID(), i.token).Send("hash/delete")
	return nil
}

func (i *hash) Container() Container {
	return i.center.container
}

func (i *hash) Instance() interface{} {
	return i.instance
}

func (i *hash) SetInstance(instance interface{}) {
	i.instance = instance
}
func (i *hash) Context() context.Context {
	return i.ctx
}

func (i *hash) Token() string {
	return i.token
}
func (i *hash) SetConfig(key NodeConfig, value interface{}) {
	if i.init {
		i.center.container.Error(key + " init finish, not allow change config")
		return
	}
	switch key {
	case ThreadSafe:
		i.SetThread(value.(bool))
	default:
		i.center.container.Error(key + " is not effect config")
	}
}

func (m *HashCenter) ID() string {
	return m.container.ID()
}

func (m *HashCenter) Emit(module string, token string, event interface{}, handler asteroid.ToCallbackHandler, param ...interface{}) {
	identity := module + "/" + token
	node := m.container.region.Alloc(identity)
	m.container.global.Direct(node).List(module, token, event, param).CallbackTo(handler).Send("hash/exec")
}

func (m *HashCenter) Exec(module string, token string, event interface{}, param ...interface{}) ([]interface{}, error) {
	identity := module + "/" + token
	r := resultPool.Get().(chan interface{})
	node := m.container.region.Alloc(identity)
	m.container.global.Direct(node).List(module, token, event, param).CallbackTo(func(success bool, data interface{}) {
		if success {
			r <- data
		} else {
			r <- fmt.Errorf("cluster fail:%s", node)
		}
	}).Send("hash/exec")
	rr := <-r
	if e, ok := rr.(error); ok {
		return nil, e
	}
	return rr.([]interface{}), nil
}

func (m *HashCenter) close() {
	// 零时任务，容许错误，不进行故障转移
	m.manages.Range(func(key, value interface{}) bool {
		manager := value.(*ServiceManager)
		manager.Services(func(key, value interface{}) bool {
			value.(*hash).close()
			return true
		})
		manager.Clients(func(key, value interface{}) bool {
			value.(*Client).close()
			return true
		})
		return true
	})
}
